package com.demo.config;

import com.demo.constant.KafkaCode;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.*;

import java.util.HashMap;
import java.util.Map;

@Configuration
@EnableKafka
public class TopicConfig {

    //  kafka管理员，委派给AdminClient以创建在应用程序上下文中定义的主题的管理员。
	@Bean
	public KafkaAdmin kafkaAdmin() {
		Map<String, Object> configs = new HashMap();
			configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaCode.BOOTSTRAP_SERVERS);
		return new KafkaAdmin(configs);
	}

    //   kafka的管理客户端，用于创建、修改、删除主题等
    @Bean
    public AdminClient adminClient() {
        return AdminClient.create(kafkaAdmin().getConfig());
    }

    //  创建新的 topinfo 的Topic，如果kafka中topinfo 的topic已经存在，则忽略。
	@Bean
	public NewTopic foo() {
        //第一个是参数是topic名字，第二个参数是分区个数，第三个是topic的复制因子个数
		//当broker个数为1个时会创建topic失败，
		//提示：replication factor: 2 larger than available brokers: 1
		//只有在集群中才能使用kafka的备份功能
		return new NewTopic("foo", 2, (short) 2);
	}
	@Bean
	public NewTopic bar() {
		return new NewTopic("bar", 2, (short) 2);
	}
	@Bean
	public NewTopic topic1(){
		return new NewTopic("topic1", 2, (short) 2);
	}
	@Bean
	public NewTopic topic2(){
		return new NewTopic("topic2", 2, (short) 2);
	}

}